package com.sca.point.utils;

import com.google.common.hash.Funnels;
import com.google.common.hash.Hashing;
import com.sca.point.constant.RedisKeyConstant;
import lombok.Getter;
import lombok.Setter;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 * 分布式布隆过滤器，若删除频繁，则需定时重建
 * 实现参考 {@link com.google.common.hash.BloomFilter}
 *
 * @author liuxianfei
 * @since 2020/12/7 11:03
 */
@Setter
@Getter
@Component
@ConfigurationProperties(prefix = "sca.bf")
public class RedisBloomFilter {

    /**
     * 预计插入数量
     */
    private int size;

    /**
     * 期望误判率
     */
    private double fpp;

    /**
     * 位数组长度
     */
    private long numBits;

    /**
     * hash函数数量
     */
    private int numHashFunctions;

    private final RedisTemplate<String, Object> redisTemplate;

    public RedisBloomFilter(RedisTemplate<String, Object> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    @PostConstruct
    public void init() {
        this.numBits = optimalNumOfBits(size, fpp);
        this.numHashFunctions = optimalNumOfHashFunctions(size, numBits);
    }

    private long optimalNumOfBits(long n, double p) {
        if (p == 0) {
            p = Double.MIN_VALUE;
        }
        return (long) (-n * Math.log(p) / (Math.log(2) * Math.log(2)));
    }

    private int optimalNumOfHashFunctions(long n, long m) {
        // (m / n) * log(2), but avoid truncation due to division!
        return Math.max(1, (int) Math.round((double) m / n * Math.log(2)));
    }

    public void put(String key) {
        long[] indexs = getIndex(key);
        redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
            connection.openPipeline();
            for (long index : indexs) {
                connection.setBit(RedisKeyConstant.POINT_BF_USER_ID.getBytes(), index, true);
            }
            connection.close();
            return null;
        });
    }

    public boolean mightContain(String key) {
        long[] indexs = getIndex(key);
        List<Object> list = redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
            connection.openPipeline();
            for (long index : indexs) {
                connection.getBit(RedisKeyConstant.POINT_BF_USER_ID.getBytes(), index);
            }
            connection.close();
            return null;
        });
        return !list.contains(false);
    }

    private long[] getIndex(String key) {
        long h1 = hash(key);
        long h2 = h1 >>> 16;
        long[] result = new long[numHashFunctions];
        for (int i = 0; i < numHashFunctions; i++) {
            long temp = h1 + i * h2;
            if (temp < 0) {
                temp = ~temp;
            }
            result[i] = temp % numBits;
        }
        return result;
    }

    private long hash(String key) {
        return Hashing.murmur3_128().hashObject(key, Funnels.stringFunnel(StandardCharsets.UTF_8)).asLong();
    }

}
